package my.thesis.bolts.fieldExtractors;

import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import my.thesis.analytics.metrics.ThroughputMetric;
import my.thesis.analytics.metrics.whereAmIMetric;

import org.joda.time.DateTime;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;
import org.json.JSONArray;
import org.json.JSONObject;
import org.json.simple.parser.JSONParser;
import org.json.simple.parser.ParseException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import backtype.storm.metric.LoggingMetricsConsumer;
import backtype.storm.metric.api.CountMetric;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;


public class TweetFieldExtractorJSON extends BaseRichBolt 
{
	//transient CountMetric _countMetric;
	private static final long serialVersionUID = 1L;	
	private OutputCollector collector;
		
	transient whereAmIMetric _whereAmI;

	public void prepare(Map stormConf, TopologyContext context,OutputCollector collector)
	{
		this.collector=collector;
		
		_whereAmI=new whereAmIMetric();
		context.registerMetric("whereAmI", _whereAmI, 45);
	}
		
	public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) 
	{		
			outputFieldsDeclarer.declareStream("hashStream", new Fields("id","millis","lang","hash"));
			outputFieldsDeclarer.declareStream("textStream", new Fields("id","millis","country","lang","text","lat","lon"));   
			outputFieldsDeclarer.declareStream("influenceStream", new Fields("id","username","mention","followers","text")); 
	}
	
	public void execute(Tuple tuple) 
	{
		//_throughputMetric.incrTuples();
		
		Set<String> hashtags = new HashSet<String>();
		Set<String> mentions = new HashSet<String>();

		String jsonString = tuple.getStringByField("jsonString");
		
		JSONObject obj = null;
		try
		{
			obj = new JSONObject(jsonString);
		}
		catch(Exception ex)
		{
			System.out.println("Something went wrong...");
		}
		
		if(obj != null)
		{
			Long id = obj.getLong("id");
			String username=obj.getString("username");
			int followers = obj.getInt("followers");
			
			String lang = obj.getString("lang");
			
			String text = obj.getString("text");
			String country = obj.getString("country");
			
			//Long millis=obj.getLong("millis");
			
			DateTime timestamp=new DateTime(obj.get("createdat"));
			Long millis = timestamp.getMillis();
			
			//TEST
			//DateTime timestamp = DateTime.now();
			//Long millis = timestamp.getMillis();
			//TEST
			
			Double  lat = obj.getDouble("lat");
			Double lon = obj.getDouble("lon");
			
			if(obj.has("hashtags"))
			{
				JSONArray hashtagsArray = obj.getJSONArray("hashtags");
				for (int i=0; i < hashtagsArray.length(); i++)
				{
					hashtags.add(hashtagsArray.getJSONObject(i).getString("text"));
				}
			}
			
			if(obj.has("mentions"))
			{
				JSONArray mentionsArray = obj.getJSONArray("mentions");
				for (int i=0; i < mentionsArray.length(); i++)
				{
					mentions.add(mentionsArray.getJSONObject(i).getString("name"));
				}
			}
			
			for (String hashtag : hashtags)
			{
				collector.emit("hashStream", new Values(id,millis,lang,hashtag));
			}
			
			collector.emit("textStream", new Values(id,millis,country,lang,text,lat,lon));
			
			for (String mention : mentions)
			{
				collector.emit("influenceStream",new Values(id,username,mention,followers,text));
			}				
		}
		else
		{
			//do nothing
		}		
	}	
}
